package streamAPI.asyncio;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.util.EntityUtils;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.function.Consumer;
import java.util.function.Supplier;

/**
 * @author yue.cao
 * @since 10-21-2020
 */
@Slf4j
public class AsyncGetWebRequest extends RichAsyncFunction<String, Tuple2<String, String>> {

	private transient CloseableHttpAsyncClient httpAsyncClient;

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		RequestConfig config = RequestConfig
				.custom()
				.setSocketTimeout(3000)
				.setConnectTimeout(3000)
				.build();
		httpAsyncClient = HttpAsyncClients.custom().setMaxConnTotal(20).setDefaultRequestConfig(config).build();
		httpAsyncClient.start();
	}


	@Override
	public void asyncInvoke(String input, ResultFuture<Tuple2<String, String>> resultFuture) throws Exception {
		HttpGet httpGet = new HttpGet("https://www.google.com/");
		Future<HttpResponse> future = httpAsyncClient.execute(httpGet, null);
		CompletableFuture.supplyAsync(new Supplier<String>() {
			@Override
			public String get() {
				try {
					HttpResponse response = future.get();
					String res = null;
					if (response.getStatusLine().getStatusCode() == 200) {
						res = EntityUtils.toString(response.getEntity());
					}
					return res;
				} catch (Exception e) {
					log.error("http error", e);
					return null;
				}

			}
		}).thenAccept(new Consumer<String>() {
			@Override
			public void accept(String res) {
				resultFuture.complete(Collections.singleton(Tuple2.of(input, res)));
			}
		});
	}

	@Override
	public void timeout(String input, ResultFuture<Tuple2<String, String>> resultFuture) throws Exception {
		log.warn("timeout input{} resultFuture:{}", resultFuture);
	}

	@Override
	public void close() throws Exception {
		super.close();
		if (httpAsyncClient != null)
			httpAsyncClient.close();
	}
}
